/**
 * 
 */
package edu.umd.clip.lm.factors;

import java.io.*;
import java.util.concurrent.locks.*;

/**
 * @author Denis Filimonov <den@cs.umd.edu>
 *
 */
public class ParallelInputParser {
	private static final int LINES_PER_BATCH = 100;

	private final Callback callback;
	private final Lock lock;
	private final int jobs;
	private boolean isDone;
	private final InputParser parser;
	private final boolean mergeLines;
	private Worker lastWorker;
	
	public interface Callback {
		void process(long[] tuples);
	}
	
	public ParallelInputParser(InputParser parser, Callback callback, int jobs, boolean mergeLines) {
		this.parser = parser;
		this.callback = callback;
		this.jobs = jobs;
		this.lock = new ReentrantLock();
		this.mergeLines = mergeLines;
	}
	
	private class Worker implements Runnable {
		private final BufferedReader input;
		private long[] nextLine;
		
		public Worker(BufferedReader input) {
			this.input = input;
		}
		/* (non-Javadoc)
		 * @see java.lang.Runnable#run()
		 */
		@Override
		public void run() {
			String buffer[] = new String[LINES_PER_BATCH];
			while(true) {
				int count = 0;
				lock.lock();
				
				Worker prevWorker = lastWorker;
				lastWorker = this;
				nextLine = null;
				
				if (!isDone) {
					for(;count < LINES_PER_BATCH; ++count) {
						String line = null;
						try {
							line = input.readLine();
						} catch (IOException e) {
						}
						if (line == null) {
							isDone = true;
							break;
						}
						buffer[count] = line;
					}
				}
				lock.unlock();
				if (count == 0) {
					if (mergeLines && prevWorker != null) {
						prevWorker.nextLine = new long[0];
						synchronized(prevWorker) {
							notifyAll();
						}
					}
					break;
				}
				
				long prevLine[] = null;
				for(int i=0; i<count; ++i) {
					long[] tuples = parser.parseSentence(buffer[i]);
					
					if (mergeLines) {
						if (prevLine == null) {
							// send to the previous worker
							if (prevWorker != null) {
								prevWorker.nextLine = tuples;
								synchronized(prevWorker) {
									prevWorker.notifyAll();
								}
							}
						} else {
							mergeLines(prevLine, tuples);
						}
					}
					callback.process(tuples);
					prevLine = tuples;
				}
				
				if (mergeLines) {
					if (prevLine != null) {
						if (nextLine == null) {
							synchronized(this) {
								// wait for the next worker to set the nextLine
								while(nextLine == null) {
									try {
										wait();
									} catch (InterruptedException e) {}
								}
							}
						}
						if (nextLine.length > 0) {
							mergeLines(prevLine, nextLine);
						}
					}
				}
			}
		}
		
		private void mergeLines(long prevLine[], long newLine[]) {
			final int numStartTuples = parser.getStartTuples().length;
			final int numEndTuples = parser.getEndTuples().length;
			
			int prevSize = Math.min(prevLine.length - numStartTuples - numEndTuples, numStartTuples-1);
			int newSize = Math.min(newLine.length - numStartTuples - numEndTuples, numStartTuples-1);
			long tuples[] = new long[prevSize + newSize];
			
			System.arraycopy(prevLine, prevLine.length - numEndTuples - prevSize, tuples, 0, prevSize);
			System.arraycopy(newLine, numStartTuples, tuples, prevSize, newSize);
			
			callback.process(tuples);
		}
	}
	
	public void parse(BufferedReader input) {
		this.isDone = false;
		this.lastWorker = null;
		
		Thread workers[] = new Thread[jobs];
		for(int i = 0; i<jobs; ++i) {
			workers[i] = new Thread(new Worker(input));
			workers[i].start();
		}
		
		try {
			for(int i = 0; i<jobs; ++i) {
				workers[i].join();
			}
		} catch (InterruptedException e) {
		}
	}
}
